package com.hkbigdata.reallywar;

import com.hkbigdata.bean.MarketingUserBehavior;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;


import java.util.Arrays;
import java.util.Random;
import java.util.List;

/**
 * @author liuanbo
 * @creat 2024-04-22-14:38
 * @see 2194550857@qq.com
 */
public class Flink03_Project_AppAnalysis_By_Chanel {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        DataStreamSource<MarketingUserBehavior> marketingUserBehaviorDataStreamSource = env.addSource(new AppMarketingDataSource());


/*        SingleOutputStreamOperator<Tuple2<String, Integer>> wordtoone = marketingUserBehaviorDataStreamSource.map((MarketingUserBehavior data) ->
                Tuple2.of(data.getBehavior() + "-" + data.getChannel(), 1)
        ).returns(Types.TUPLE(Types.STRING, Types.INT));*/
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordtoone = marketingUserBehaviorDataStreamSource.map((MarketingUserBehavior data) ->
                Tuple2.of(data.getBehavior() + "-" + data.getChannel(), 1)
        ).returns(Types.TUPLE(Types.STRING, Types.INT));

        wordtoone.keyBy(data -> data.f0)
                .sum(1)
                .print();

        env.execute();

    }

    public static class AppMarketingDataSource extends RichSourceFunction<MarketingUserBehavior> {
        boolean scanrun = true;
        Random random = new Random();
        List<String> behavior = Arrays.asList("download", "install", "update", "uninstall");
        List<String> channel = Arrays.asList("xiaomi", "apple", "huawei", "qq", "baidu", "google");

        @Override

        public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
            while (scanrun) {
                MarketingUserBehavior marketingUserBehavior = new MarketingUserBehavior(
                        (long) random.nextInt(100000),
                        behavior.get(random.nextInt(behavior.size())),
                        channel.get(random.nextInt(channel.size())),
                        System.currentTimeMillis()
                );
                ctx.collect(marketingUserBehavior);
                Thread.sleep(2000);
            }
        }

        @Override
        public void cancel() {
            scanrun = false;
        }
    }
}
